
import com.nekonomics.lakehouse.common.util.SqlUtil._
import com.nekonomics.lakehouse.common.FlinkRunner._


/**
 * 粒度为order item
 * 度量值为数量、商品单价、销售金额
 */

object OrderItemDetailDwd {

  def main(args: Array[String]): Unit = {

    implicit val config: SqlRunnerConfig = SqlRunnerConfig(groupId = "order_item_detail_dwd", port = 8083, ckp = true)

    runSql { (_, tableEnv) =>
      tableEnv.executeSql(
        sql"""
             |CREATE TABLE IF NOT EXISTS dwd.order_item_detail_dwd (
             |  order_item_id BIGINT,
             |  order_id BIGINT,
             |  sku_id BIGINT,
             |  spu_id BIGINT,
             |  merchant_id BIGINT,
             |  user_id BIGINT,
             |  quantity INT,
             |  price DECIMAL(10, 2),
             |  total_amount DECIMAL(10, 2),
             |  order_status STRING,
             |  order_created_at TIMESTAMP(3),
             |  payment_status STRING,
             |  paid_at TIMESTAMP(3),
             |  PRIMARY KEY (order_item_id) NOT ENFORCED
             |);
             |""".stripMargin
      )

      tableEnv.executeSql(
        sql"""
             |INSERT INTO dwd.order_item_detail_dwd
             |SELECT
             |  oi.id,
             |  oi.order_id,
             |  oi.sku_id,
             |  s.spu_id,
             |  spu.merchant_id,
             |  o.user_id,
             |  oi.quantity,
             |  oi.price_at_purchase AS price,
             |  oi.price_at_purchase * oi.quantity AS total_amount,
             |  o.status AS order_status,
             |  o.created_at AS order_created_at,
             |  p.status AS payment_status,
             |  p.paid_at
             |FROM ods.order_item_ods oi
             |JOIN ods.order_ods o ON oi.order_id = o.id
             |LEFT JOIN ods.payment_ods p ON p.order_id = o.id
             |JOIN ods.sku_ods s ON oi.sku_id = s.id
             |JOIN ods.spu_ods spu ON s.spu_id = spu.id
             |WHERE
             |  oi.is_deleted = false
             |  AND o.is_deleted = false
             |  AND s.is_deleted = false
             |  AND spu.is_deleted = false;
             |""".stripMargin
      )
    }
  }
}

